package com.atguigu.flink.state.keyedstate;

import com.atguigu.flink.func.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Created by 黄凯 on 2023/6/20 0020 20:39
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 针对每种传感器输出最高的3个水位值。
 * *
 * *          上午的ListState：
 * *                  MapTask: 收到了数据 s1,s2
 * *                      只有一个ListState。 当前Task收到的所有数据共用同一个ListState。
 * *
 * *          下午的ListState是KeyedState:
 * *                  MapTask: 收到了数据 s1(key:s1),s2(key:s2)
 * *                      每个key都有自己的ListState。
 * *                          所有的key为s1的数据，都用一个ListState
 * *                          所有的key为s2的数据，都用另一个ListState
 */
public class Flink06_ListState {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> ds = env
                .socketTextStream("127.0.0.1", 8888)
                .map(new WaterSensorMapFunction());

        KeyedStream<WaterSensor, String> keyedDS = ds.keyBy(WaterSensor::getId);

        keyedDS.process(
                new KeyedProcessFunction<String, WaterSensor, String>() {

                    /*
                    ListState.clear(): 清空状态中的值。
                    添加:
                    listState.add();
                    listState.addAll();
                    查询：
                    listState.get()
                    覆盖写:
                         listState.update(); 等价于，先清空，再写入

                 */
                    private ListState<Integer> listState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        //找到状态的备份存储,取出状态并赋值
                        listState = getRuntimeContext().getListState(new ListStateDescriptor<>("top3", Integer.class));

                    }

                    @Override
                    public void processElement(WaterSensor value,
                                               KeyedProcessFunction<String, WaterSensor, String>.Context ctx,
                                               Collector<String> out) throws Exception {

                        //加入集合
                        listState.add(value.getVc());

                        //排序比较取前3。 Stream
                        List<Integer> top3 = StreamSupport
                                .stream(listState.get().spliterator(), true)
                                .sorted(Comparator.reverseOrder())
                                .limit(3)
                                .collect(Collectors.toList());

                        out.collect(ctx.getCurrentKey() + " 目前的前3vc是:" + top3);

                        //始终只留当前的前三名
                        listState.update(top3);

                    }
                }
        ).print();

        env.execute();

    }

}
